{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[查看原文](https://mp.weixin.qq.com/s?subscene=23&__biz=MzIxMjY0NzA0Mg==&mid=2247484497&idx=1&sn=73b5a4a16d47052d8387288b15c4475c&chksm=9743a9bca03420aa9cdf7d50d3d2d1ada8b5e6d648ae1789c3191f82a4cf4c2e93742f1947c3&scene=7&key=792145416790f24e440f7da350a65b1043121a648dddda4b08712a81e3871c962c474b251d300a9e15f8835243b9460d610a9c1f9ecd1df964d93a3dee3fa0310ecf81dc20d55c146e5c450b2c65c3d1&ascene=0&uin=MTk3NzE3NDc2Nw%3D%3D&devicetype=Windows+10&version=62080079&lang=zh_CN&exportkey=AZ%2F%2FJdULTeZCkcbhsIEibqc%3D&pass_ticket=%2Bc9tRFUl4AHcIKDI7n6ZxyHigHsPZHZoL1Txx8DYh1bkVCqrd7KY4laIwGbeApgQ)\n",
    "\n",
    "简单记录一下使用 FastAPI 完成对数据库的 CRUD 操作。在参考文档的基础上，增加了 U、D 部分。\n",
    "\n",
    "* https://fastapi.tiangolo.com/tutorial/sql-databases/\n",
    "    \n",
    "# 安装依赖\n",
    "\n",
    "```sh\n",
    "pip install fastapi\n",
    "pip install uvicorn\n",
    "pip install sqlalchemy\n",
    "```\n",
    "\n",
    "# 2. 目录结构\n",
    "\n",
    "![pic](./assets/fastapi1.webp)\n",
    "\n",
    "其中：\n",
    "\n",
    "*   crud.py             完成对数据库的 CRUD 操作\n",
    "    \n",
    "*   database.py     关于数据库相关配置\n",
    "    \n",
    "*   main.py             fastapi 路由部分\n",
    "    \n",
    "*   models.py         定义表结构\n",
    "    \n",
    "*   schemas.py      定义 pydantic models 信息，也就是 schemas\n",
    "    \n",
    "*   \\__init__.py  \n",
    "    \n",
    "\n",
    "# 3. 详细设计\n",
    "\n",
    "## 3.1 database.py  \n",
    "\n",
    "首先需要定义数据库部分，作为 demo，使用 sqlite 作为我们的数据库。这部分都是常规配置操作。\n",
    "\n",
    "```py\n",
    "from sqlalchemy import create_engine\n",
    "from sqlalchemy.ext.declarative import declarative_base\n",
    "from sqlalchemy.orm import sessionmaker\n",
    "SQLALCHEMY_DATABASE_URL = \"sqlite:///./test.db\"\n",
    "engine = create_engine(\n",
    "    SQLALCHEMY_DATABASE_URL, connect_args={\"check_same_thread\": False}\n",
    ")\n",
    "SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)\n",
    "Base = declarative_base()\n",
    "```\n",
    "\n",
    "## 3.2 models.py\n",
    "\n",
    "第二步需要定义我们的表结构，这里定义了 User 表存储用户数据，item 表用于存储物品数据，两者关系是一对多。  \n",
    "\n",
    "```py\n",
    "from sqlalchemy import Boolean, Column, ForeignKey, Integer, String\n",
    "from sqlalchemy.orm import relationship\n",
    "from .database import Base\n",
    "class User(Base):\n",
    "    __tablename__ = \"users\"\n",
    "    id = Column(Integer, primary_key=True, index=True)\n",
    "    email = Column(String, unique=True, index=True)\n",
    "    hashed_password = Column(String)\n",
    "    is_active = Column(Boolean, default=True)\n",
    "    items = relationship(\"Item\", back_populates=\"owner\")\n",
    "    def to_dict(self):\n",
    "        return {c.name: getattr(self, c.name) for c in self.__table__.columns}\n",
    "class Item(Base):\n",
    "    __tablename__ = \"items\"\n",
    "    id = Column(Integer, primary_key=True, index=True)\n",
    "    title = Column(String, index=True)\n",
    "    description = Column(String, index=True)\n",
    "    owner_id = Column(Integer, ForeignKey(\"users.id\"))\n",
    "    owner = relationship(\"User\", back_populates=\"items\")\n",
    "    def to_dict(self):\n",
    "        return {c.name: getattr(self, c.name) for c in self.__table__.columns}\n",
    "```\n",
    "\n",
    "## 3.3 schemas.py\n",
    "\n",
    "第三步需要定义fastapi 中 schemas 信息，后续路由、CRUD 时都需要使用。由于增删改查需要不同的schema，所以官网的最佳实践一般都是通过继承解决。\n",
    "\n",
    "```py\n",
    "from typing import List\n",
    "from pydantic import BaseModel\n",
    "class ItemBase(BaseModel):\n",
    "    title: str\n",
    "    description: str = None\n",
    "class ItemCreate(ItemBase):\n",
    "    pass\n",
    "class ItemUpdate(ItemBase):\n",
    "    pass\n",
    "class Item(ItemBase):\n",
    "    id: int\n",
    "    owner_id: int\n",
    "    class Config:\n",
    "        orm_mode = True\n",
    "class UserBase(BaseModel):\n",
    "    email: str\n",
    "class UserCreate(UserBase):\n",
    "    password: str\n",
    "class UserUpdate(UserBase):\n",
    "    is_active: bool\n",
    "class User(UserBase):\n",
    "    id: int\n",
    "    is_active: bool\n",
    "    items: List[Item] = []\n",
    "    class Config:\n",
    "        orm_mode = True\n",
    "```\n",
    "\n",
    "## 3.4 crud.py\n",
    "\n",
    "接下来就是实际完成 CRUD，官网上没有对 update、delete 做实例展示，这里补充一下。  \n",
    "\n",
    "```py\n",
    "from sqlalchemy.orm import Session\n",
    "from . import models, schemas\n",
    "def get_user(db: Session, user_id: int):\n",
    "    return db.query(models.User).filter(models.User.id == user_id).first()\n",
    "def get_user_by_email(db: Session, email: str):\n",
    "    return db.query(models.User).filter(models.User.email == email).first()\n",
    "def get_users(db: Session, skip: int = 0, limit: int = 100):\n",
    "    return db.query(models.User).offset(skip).limit(limit).all()\n",
    "def create_user(db: Session, user: schemas.UserCreate):\n",
    "    fake_hashed_password = user.password + \"notreallyhashed\"\n",
    "    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)\n",
    "    db.add(db_user)\n",
    "    db.commit()\n",
    "    db.refresh(db_user)\n",
    "    return db_user\n",
    "def update_user(db: Session, user_id: int, update_user: schemas.UserUpdate):\n",
    "    db_user = db.query(models.User).filter(models.User.id == user_id).first()\n",
    "    if db_user:\n",
    "        update_dict = update_user.dict(exclude_unset=True)\n",
    "        for k, v in update_dict.items():\n",
    "            setattr(db_user, k, v)\n",
    "        db.commit()\n",
    "        db.flush()\n",
    "        db.refresh(db_user)\n",
    "        return db_user\n",
    "def delete_user(db: Session, user_id: int):\n",
    "    db_user = db.query(models.User).filter(models.User.id == user_id).first()\n",
    "    if db_user:\n",
    "        db.delete(db_user)\n",
    "        db.commit()\n",
    "        db.flush()\n",
    "        return db_user\n",
    "def get_items(db: Session, skip: int = 0, limit: int = 100):\n",
    "    return db.query(models.Item).offset(skip).limit(limit).all()\n",
    "def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):\n",
    "    db_item = models.Item(**item.dict(), owner_id=user_id)\n",
    "    db.add(db_item)\n",
    "    db.commit()\n",
    "    db.refresh(db_item)\n",
    "    return db_item\n",
    "def relate_user_item(db: Session, user_id: int, item_id: int):\n",
    "    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()\n",
    "    if db_item:\n",
    "        db_item.owner_id = user_id\n",
    "        db.commit()\n",
    "        db.flush()\n",
    "        return db.query(models.User).filter(models.User.id == user_id).first()\n",
    "def update_item(db: Session, item_id: int, update_item: schemas.ItemUpdate):\n",
    "    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()\n",
    "    if db_item:\n",
    "        update_dict = update_item.dict(exclude_unset=True)\n",
    "        for k, v in update_dict.items():\n",
    "            setattr(db_item, k, v)\n",
    "        db.commit()\n",
    "        db.flush()\n",
    "        db.refresh(db_item)\n",
    "        return db_item\n",
    "def delete_item(db: Session, item_id: int):\n",
    "    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()\n",
    "    if db_item:\n",
    "        db.delete(db_item)\n",
    "        db.commit()\n",
    "        db.flush()\n",
    "        return db_item\n",
    "```\n",
    "\n",
    "## 3.5 main.py\n",
    "\n",
    "最后完成路由的配置，基本都是按照官网上来的，只是稍微补充了一些删和改的内容，为了让这个 demo 更加完善。  \n",
    "\n",
    "```py\n",
    "from typing import List\n",
    "from fastapi import Depends, FastAPI, HTTPException\n",
    "from sqlalchemy.orm import Session\n",
    "from . import crud, models, schemas\n",
    "from .database import SessionLocal, engine\n",
    "models.Base.metadata.create_all(bind=engine)\n",
    "app = FastAPI()\n",
    "# Dependency\n",
    "def get_db():\n",
    "    try:\n",
    "        db = SessionLocal()\n",
    "        yield db\n",
    "    finally:\n",
    "        db.close()\n",
    "@app.post(\"/users/\", response_model=schemas.User)\n",
    "def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):\n",
    "    db_user = crud.get_user_by_email(db, email=user.email)\n",
    "    if db_user:\n",
    "        raise HTTPException(status_code=400, detail=\"Email already registered\")\n",
    "    return crud.create_user(db=db, user=user)\n",
    "@app.get(\"/users/\", response_model=List[schemas.User])\n",
    "def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):\n",
    "    users = crud.get_users(db, skip=skip, limit=limit)\n",
    "    return users\n",
    "@app.get(\"/users/{user_id}\", response_model=schemas.User)\n",
    "def read_user(user_id: int, db: Session = Depends(get_db)):\n",
    "    db_user = crud.get_user(db, user_id=user_id)\n",
    "    if db_user is None:\n",
    "        raise HTTPException(status_code=404, detail=\"User not found\")\n",
    "    return db_user\n",
    "@app.delete('/users/{user_id}', response_model=schemas.User)\n",
    "def delete_user(user_id: int, db: Session = Depends(get_db)):\n",
    "    db_user = crud.delete_user(db, user_id=user_id)\n",
    "    if db_user is None:\n",
    "        raise HTTPException(status_code=404, detail=\"User not found\")\n",
    "    return db_user\n",
    "@app.put(\"/users/{user_id}\", response_model=schemas.User)\n",
    "def update_user(user_id: int, update_user: schemas.UserUpdate, db: Session = Depends(get_db)):\n",
    "    updated_user = crud.update_user(db, user_id, update_user)\n",
    "    if updated_user is None:\n",
    "        raise HTTPException(status_code=404, detail=\"User not found\")\n",
    "    return updated_user\n",
    "@app.post(\"/users/{user_id}/items/\", response_model=schemas.Item)\n",
    "def create_item_for_user(\n",
    "    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)\n",
    "):\n",
    "    return crud.create_user_item(db=db, item=item, user_id=user_id)\n",
    "@app.get(\"/items/\", response_model=List[schemas.Item])\n",
    "def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):\n",
    "    items = crud.get_items(db, skip=skip, limit=limit)\n",
    "    return items\n",
    "@app.put(\"/items/{user_id}/{item_id}/\", response_model=schemas.User)\n",
    "def relate_user_item(user_id: int, item_id: int, db: Session = Depends(get_db)):\n",
    "    user = crud.relate_user_item(db=db, item_id=item_id, user_id=user_id)\n",
    "    return user\n",
    "@app.put(\"/items/{item_id}\", response_model=schemas.Item)\n",
    "def update_item(item_id: int, update_item: schemas.ItemUpdate, db: Session = Depends(get_db)):\n",
    "    updated_item = crud.update_item(db, item_id, update_item)\n",
    "    if updated_item is None:\n",
    "        raise HTTPException(status_code=404, detail=\"Item not found\")\n",
    "    return updated_item\n",
    "@app.delete('/items/{item_id}', response_model=schemas.Item)\n",
    "def delete_item(item_id: int, db: Session = Depends(get_db)):\n",
    "    db_item = crud.delete_item(db, item_id=item_id)\n",
    "    if db_item is None:\n",
    "        raise HTTPException(status_code=404, detail=\"User not found\")\n",
    "    return db_item\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 4. Tests\n",
    "\n",
    "在文件路径下执行，--reload 当代码有修改时，可以自动加载。\n",
    "\n",
    "```sh\n",
    "uvicorn sql_app.main:app --reload\n",
    "```\n",
    "\n",
    "可以通过 postman 发送请求做测试，这里贴图一张。\n",
    "\n",
    "<img src=\"./assets/fastapi2.jpg\" width=70%>\n",
    "最后 fastapi 还可以自动生成文档，可以访问 http://127.0.0.1:8000/docs 查阅。"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.3"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {
    "height": "calc(100% - 180px)",
    "left": "10px",
    "top": "150px",
    "width": "186.818px"
   },
   "toc_section_display": true,
   "toc_window_display": true
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
